/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.ram;

import org.sabayframework.channel.Lock;
import org.sabayframework.channel.impl.AbstractChannel;


/**
 * Single-threaded implementation of Channel.
 * Lock size can't exceed size of the array.
 * @author Alex Antonau
 * */
public class ArrayChannel<T> extends AbstractChannel<T> {

	private final T[] items;
    /** items index for next take, poll or remove */
    private int takeIndex;
    /** items index for next put, offer, or add. */
    private int putIndex;
    /** Number of items in the queue */
    private int count;
    /** closed flag */
    private boolean closed;

    private ReadLock readLock;
    private WriteLock writeLock;
    
    /**
     * Creates channel from given array of elements.
     * @SafeVarargs 
     * */
	public ArrayChannel(T... items) {
		super();
		this.items = items;
		this.count = items.length;
	}

    /**
     * Creates empty array channel
     * */
	@SuppressWarnings("unchecked")
	public ArrayChannel(int size) {
		this.items = (T[])new Object[size];
	}
    /**
     * Creates array channel with a minimum size and given array of elements. 
     * */
	@SuppressWarnings("unchecked")
	public ArrayChannel(int size, Object... items) {
		size = Math.max(size, items.length);
		this.items = (T[])new Object[size];
		System.arraycopy(items, 0, this.items, 0, items.length);
		this.count = items.length;
		this.putIndex = count % this.items.length;
	}

	private final class ReadLock implements Lock<T>{
    	private int locked=-1;
		
		public boolean tryLock(final int size) {
			if(locked != -1) throw new IllegalStateException("Lock is already acquired");
			if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
			if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
			
			if(size <= count){
				locked = size;
				count-=size;
				return true;
			}
			if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
			return false;
		}
		@Override
		public int size() {
			return locked;
		}

		@Override
		public T get(final int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (takeIndex + index) % items.length;
			return items[i];
		}
		@Override
		public void set(final int index, final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (takeIndex + index) % items.length;
			items[i] = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			for(int i=0;i<locked;++i){
				res[i] = items[(takeIndex+i)%items.length];
			}
			return res;
		}
		@Override
		public void unlock() {
			if(locked!=-1){
				for(int i=0;i<locked;++i)
					items[(takeIndex+i)%items.length] = null;
				takeIndex = (takeIndex+locked) % items.length;
				locked = -1;
			}
		}
		@Override
		public void cancel() {
			if(locked!=-1){
				count+=locked;
				locked = -1;
			}
		}
		
    }
    
	@Override
	public Lock<T> getReadLock(final int size) {
		if(readLock == null) readLock = new ReadLock();
		return readLock.tryLock(size)? readLock:null;
	}

	private final class WriteLock implements Lock<T>{
		private int locked=-1;
		
		public boolean tryLock(final int size) {
			if(locked != -1) throw new IllegalStateException("Lock is already acquired");
			if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
			if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
			if(size <= freeSize()){
				locked = size;
				count+=size;
				return true;
			}
			if(size > items.length) throw new IllegalArgumentException(String.format("Lock size %d exceeds array size %d", size, items.length));
			return false;
		}
		@Override
		public int size() {
			return locked;
		}
		@Override
		public T get(final int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (putIndex + index) % items.length;
			return items[i];
		}
		@Override
		public void set(final int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			int i = (putIndex + index) % items.length;
			items[i] = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			for(int i=0;i<locked;++i){
				res[i] = items[(putIndex+i)%items.length];
			}
			return res;
		}

		@Override
		public void unlock() {
			if(locked!=-1){
				putIndex = (putIndex+locked) % items.length;
				locked = -1;
			}
		}
		@Override
		public void cancel() {
			if(locked!=-1){
				for(int i=0;i<locked;++i){
					items[(putIndex+i)%items.length] = null;
				}
				count-=locked;
				locked = -1;
			}
		}
		
	}
	@Override
	public Lock<T> getWriteLock(final int size) {
		if(writeLock == null) writeLock = new WriteLock();
		return writeLock.tryLock(size)? writeLock:null;
	}

	@Override
	public boolean isEmpty() {
		return count == 0;
	}

	@Override
	public boolean isFull() {
		return count == items.length;
	}

	@Override
	public int size() {
		return count;
	}

	@Override
	public int capacity() {
		return items.length;
	}

	@Override
	public int freeSize() {
		if(closed)return 0;
		return items.length-count;
	}

	@Override
	public void close() {
		closed=true;
	}

	@Override
	public boolean isClosed() {
		return closed;
	}
	
	@Override
	public Lock<T> waitForReadLock(int count, long timeout) {
		Lock<T> lock = getReadLock(count);
		if(lock!=null) return lock;
		throw new IllegalStateException("Wait is not implemented. Use concurrent channel instead.");
	}

	/**
	 * Returns elements are currently in the channel.
	 * */
	public T[] toArray() {
		@SuppressWarnings("unchecked")
		T[] res = (T[])new Object[count];
		for(int i=0;i<count;++i){
			res[i] = items[(takeIndex+i)%items.length];
		}
		return res;
	}

	@Override
	public T remove() throws IllegalStateException {
		if(count>=1){
			--count;
			final int i = takeIndex;
			takeIndex = (i+1) % items.length;
			final T item = items[i];
			items[i] = null;
			return item;
		}
		throw new IllegalStateException("Channel is empty");
	}

	@Override
	public void add(T elem) throws IllegalStateException {
		if(freeSize()>0){
			++count;
			final int i = putIndex;
			putIndex = (i+1) % items.length;
			items[i] = elem;
		}else 
			throw new IllegalStateException("Channel is full");
	}

}
